package bigbeard.tools.net.manager;
/**
 * @author bigbeard on 2022-09-29
 */

import bigbeard.tools.net.api.DataArrivalCallback;
import bigbeard.tools.net.api.StateChangeCallback;
import bigbeard.tools.net.api.server.IConnector;
import bigbeard.tools.net.api.server.IServer;
import bigbeard.tools.net.server.NetServerFactory;
import bigbeard.tools.thread.pools.ThreadPoolFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.java_websocket.util.NamedThreadFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * @author bigbeard on 2022-09-29
 */
public class NetServerManager {
    private final Log logger = LogFactory.getLog(NetServerManager.class);

    private final Map<String, IConnector> connectors = Collections.synchronizedMap(new HashMap<>(16));

    private ExecutorService executorService;
    private final ThreadFactory threadFactory = new NamedThreadFactory("heart-thread-%s");
    private ScheduledThreadPoolExecutor executor;

    private DataArrivalCallback dataArrivalCallback;
    private StateChangeCallback stateChangeCallback;
    private Consumer<String> consumer;

    private IServer server;

    private int heartInterval = 0;
    private boolean running = false;

    /**
     * 启动服务
     *
     * @param protocol 服务侦听协议,支持,ws(websocket),tcp,udp
     * @param port     端口
     * @throws Exception 异常
     */
    public void start(String protocol, int port) throws Exception {
        server = NetServerFactory.create(protocol);
        if (null == server) {
            throw new Exception("协议错误，当前仅仅支持 ws,tcp,udp");
        }
        executorService = ThreadPoolFactory.createNotifyThreadPool("Notify-thread-%s");
        server.onDataArrival((key, data) -> {
            logger.info("收到[" + key + "]的数据，长度:" + data.length);
            IConnector client = this.connectors.getOrDefault(key, null);
            if (null != client) {
                client.updateLastMsgTime();
            }
            if (null != this.dataArrivalCallback) {
                this.dataArrivalCallback.callback(key, data);
            }
        });
        server.onStateChange(((key, state, info) -> {
            IConnector orDefault = this.connectors.getOrDefault(key, null);
            if (null != orDefault) {
                orDefault.close();
            }
            this.connectors.remove(key);
            notifyStateChange(key, state, info);
        }));
        server.onClientAccept(((key, connector) -> {
            IConnector tmp = this.connectors.getOrDefault(key, null);
            if (null != tmp) {
                tmp.close();
            }
            connectors.remove(key);
            connectors.put(key, connector);
            if (null != this.consumer) {
                executorService.execute(() -> this.consumer.accept(key));
            }
        }));
        server.start(port);
    }

    /**
     * 停止服务
     */
    public void stop() {
        running = false;
        if (null != server) {
            server.stop();
        }
        ScheduledThreadPoolExecutor shutdownThread =
                new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("stop"));

        shutdownThread.schedule(() -> {
            if (null != executor) {
                executor.shutdown();
                executor = null;
            }
            executorService.shutdownNow();
            shutdownThread.shutdown();
        }, 5, TimeUnit.SECONDS);

    }

    public synchronized void heart(int interval) {
        if (interval <= 0) {
            interval = 0;
        }
        heartInterval = interval * 1000;
        if (!running && heartInterval > 0) {
            startHeart();
        }
    }

    public int send(String key, byte[] data) throws Exception {
        if (this.connectors.containsKey(key)) {
            IConnector connector = this.connectors.get(key);
            if (null != connector) {
                return connector.send(data);
            }
        }
        return -1;
    }

    public void onClientAccept(Consumer<String> consumer) {
        this.consumer = consumer;
    }

    public void onDataArrival(DataArrivalCallback callback) {
        this.dataArrivalCallback = callback;
    }

    public void onStateChange(StateChangeCallback callback) {
        this.stateChangeCallback = callback;
    }


    private void notifyStateChange(String key, int state, String info) {
        if (null != this.stateChangeCallback) {
            executorService.execute(() -> this.stateChangeCallback.notifyStateChange(key, state, info));
        }
    }

    private void startHeart() {
        if (running) {
            return;
        }
        running = true;
        if (null != executor) {
            try {
                executor.shutdown();
                Thread.sleep(1000);
            } catch (Exception ignored) {
            } finally {
                executor = null;
            }
        }
        int delay = 1;
        executor = new ScheduledThreadPoolExecutor(5, threadFactory);
        executor.schedule(this::executeHeart, delay, TimeUnit.SECONDS);
    }

    private synchronized void executeHeart() {
        if (heartInterval <= 0) {
            return;
        }
        Map<String, IConnector> tmp = new HashMap<>(connectors);
        tmp.forEach((k, v) -> {
            if (null != v && heartInterval > 0 &&
                System.currentTimeMillis() - v.getLastMsgTime() > heartInterval) {
                notifyStateChange(k, StateChangeCallback.CLIENT_HEART_LAST, "客户端心跳丢失");
                v.close();
            }
        });
    }
}
